package org.apache.zookeeper.server.paxos2;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.TreeMap;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

public class MetaFile implements Iterable<Entry<Long, byte[]>> {

    private final String path;
    private final boolean readonly;
    private RandomAccessFile file;
    private MetaIndex index;
    private static final Logger LOG = Logger.getLogger(MetaFile.class);

    private class MetaEntry implements Entry<Long, byte[]> {

        private long key;
        private byte[] value;

        public MetaEntry(long key, byte[] value) {
            this.key = key;
            this.value = value;
        }

        public Long getKey() {
            return key;
        }

        public byte[] getValue() {
            return value;
        }

        public byte[] setValue(byte[] value) {
            throw new UnsupportedOperationException();
        }

    }

    public MetaFile(String path, boolean readonly) {
        this.path = path;
        this.readonly = readonly;
        try {
            if (readonly)
                file = new RandomAccessFile(path, "r");
            else
                file = new RandomAccessFile(path, "rw");
        }
        catch (IOException e) {
	    LOG.error("could not read metafile", e);
            //e.printStackTrace();
            System.exit(-1);
        }
        index = new MetaIndex(path + ".index", readonly);

    }

    public String getPath() {
        return this.path;
    }

    private byte[] data = new byte[(8 + BlockMeta.BLOCK_META_SIZE) * 65536];
    private ByteBuffer buffer = ByteBuffer.wrap(data);

    public void write(long key, byte[] value) {
        if (readonly)
            throw new RuntimeException("Cannot write to a readonly metafile");
        index.write(key);
        try {
            if (buffer.remaining() == 0) {
                file.write(data);
                buffer.rewind();
            }
            buffer.putLong(key);
            buffer.put(value);
        }
        catch (IOException e) {
	    LOG.error("could not write metafile", e);
            //e.printStackTrace();
            System.exit(-1);
        }
    }

    public void close() {
        try {
            if (!readonly) {
                file.write(data, 0, buffer.position());
                index.flush();
            }
            file.close();

        }
        catch (IOException e) {
	    LOG.error("could not close metafile", e);
            //e.printStackTrace();
            System.exit(-1);
        }
    }

    public byte[] read(long key) {
        if (!readonly)
            throw new RuntimeException("Cannot read to a writeonly metafile");
        if (!index.isIn())
            return null;
        int position = index.getStartIndex(key);
        //System.out.println(path+" position="+position+" key="+key);
        if (position == -1)
            return null;
        byte[] dataToSearch = new byte[(8 + BlockMeta.BLOCK_META_SIZE) * MetaIndex.gap];
        long offset = (8 + BlockMeta.BLOCK_META_SIZE) * position * MetaIndex.gap;
        try {
            file.seek(offset);
            int size = file.read(dataToSearch);
	    if(size<=0){
		System.out.println("read return "+size);
		return null;
	    }
            ByteBuffer b = ByteBuffer.wrap(dataToSearch, 0, size);
            long searchKey;
            byte[] searchValue = new byte[BlockMeta.BLOCK_META_SIZE];
            while (b.remaining() > 0) {
                searchKey = b.getLong();
		//System.out.println("searchKey="+searchKey);
                b.get(searchValue);
                if (searchKey == key)
                    return searchValue;
                if (searchKey > key)
                    return null;
            }
            return null;
        }
        catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }

    public Iterator<Entry<Long, byte[]>> iterator() {
        return new MetaFileIterator();
    }

    public class MetaFileIterator implements Iterator<Entry<Long, byte[]>> {

        private FileInputStream fis;
        private byte[] data = new byte[(8 + BlockMeta.BLOCK_META_SIZE) * 65536];
        private ByteBuffer buffer = ByteBuffer.wrap(data);


        public MetaFileIterator() {
            try {
                fis = new FileInputStream(path);
                buffer.position(data.length);
            }
            catch (IOException e) {
		LOG.error("could not iterate file", e);
                //e.printStackTrace();
                System.exit(-1);
            }
        }

        public boolean hasNext() {
            try {
                return buffer.remaining() > 0 || (fis != null && fis.available() > 0);
            }
            catch (IOException e) {
                e.printStackTrace();
                return false;
            }
        }

        public Entry<Long, byte[]> next() {
            try {
                if (buffer.remaining() == 0) {
                    int size = fis.read(data);
                    if (size < data.length) {
                        fis.close();
                        fis = null;
                    }
                    buffer = ByteBuffer.wrap(data, 0, size);
                }
                long key = buffer.getLong();
                byte[] value = new byte[BlockMeta.BLOCK_META_SIZE];
                buffer.get(value);
                return new MetaEntry(key, value);
            }
            catch (IOException e) {
                e.printStackTrace();
                return null;
            }

        }

        public void remove() {
            throw new RuntimeException("Not implemented");
        }

    }


    @Override
    public String toString() {
        return path;
    }

    public static boolean checkMetaFileConsistency(String fileName) {
        File metaFile = new File(fileName);
        File metaIndexFile = new File(fileName + ".index");
	System.out.println("check "+metaFile+" "+metaIndexFile);
        if ((metaFile.length() / (8 + BlockMeta.BLOCK_META_SIZE) + MetaIndex.gap - 1) / MetaIndex.gap == metaIndexFile.length() / 8)
            return true;
        else
            return false;
    }


}
